package com.distributer.message.kafka;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;

import com.distributer.message.service.MessageService;

@SuppressWarnings("rawtypes")
@Component
public class KafkaSendResultHandler implements ProducerListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSendResultHandler.class);

    @Autowired
    private MessageService messageService;
    
	@Override
	public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
		LOGGER.info("消息编号： "+ key+"发送消息成功！");
		//更新消息已經成功发送
		messageService.queryMessageByMessageSquece((String)key);
	}

	@Override
	public void onError(String topic, Integer partition, Object key, Object value, Exception exception) {
		LOGGER.info("消息编号： "+ key+"发送消息失败！");
		//更新消息发送失败，需要重新发送
		messageService.queryMessageByMessageSquece((String)key);
	}

	@Override
	public boolean isInterestedInSuccess() {
		return false;
	}

}
